package com.bwie.service.Impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;

import com.bwie.pojo.TUser;
import com.bwie.service.FeignOrderService;
import com.bwie.state.OrderEnum;
import com.bwie.utils.MobileMessageUtil;
import com.bwie.utils.R;
import com.bwie.vo.PageInfoVo;


import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.mail.javamail.JavaMailSender;

import org.springframework.mail.javamail.MimeMessageHelper;
import org.springframework.stereotype.Service;
import org.springframework.beans.factory.annotation.Autowired;
import java.util.List;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.bwie.mapper.OrderMapper;
import com.bwie.pojo.Order;
import com.bwie.service.OrderService;
import org.springframework.transaction.annotation.Transactional;

import javax.mail.MessagingException;
import javax.mail.internet.MimeMessage;

@Service
@Transactional
public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements OrderService{
    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    private KafkaTemplate kafkaTemplate;
    @Autowired
    private FeignOrderService feignOrderService;
    @Autowired
    private JavaMailSender javaMailSender;
    @Override
    public R orderList(PageInfoVo pageInfoVo) {
        Page<Order> orderPage = new Page<>(pageInfoVo.getPageNum(),pageInfoVo.getPageSize());
        QueryWrapper<Order> orderQueryWrapper = new QueryWrapper<>();
        LambdaQueryWrapper<Order> orderLambdaQueryWrapper = orderQueryWrapper.lambda().orderByDesc(Order::getCreateTime);
        if(ObjectUtils.isNotEmpty(pageInfoVo.getKeyWord())){
            orderLambdaQueryWrapper.or().like(Order::getOrderName,pageInfoVo.getKeyWord());
        }
        if(ObjectUtils.isNotEmpty(pageInfoVo.getBeginTime()) && ObjectUtils.isNotEmpty(pageInfoVo.getEndTime())){
            orderLambdaQueryWrapper.or().between(Order::getCreateTime,pageInfoVo.getBeginTime(),pageInfoVo.getEndTime());
        }
        Page<Order> orderPage1 = orderMapper.selectPage(orderPage, orderLambdaQueryWrapper);
        return R.OK(orderPage1);
    }

    @Override
    public R updateOrder(Order order) {
        Order order1 = orderMapper.selectById(order.getId());
        if(ObjectUtils.isEmpty(order1)){
            return R.ERROR(500,"订单不存在");
        }
        if(order.getOrderStatus().equals(OrderEnum.ORDER_STATUS_NO.getCode())){
            if(StringUtils.isEmpty(order.getContent())){
                return R.ERROR(500,"取消订单必须填写取消原因");
            }
        }
        order1.setOrderStatus(order.getOrderStatus());
        order1.setContent(order.getContent());
        orderMapper.updateById(order1);
        /**
         * 生产者
         */
        String jsonString = JSON.toJSONString(order1);
        kafkaTemplate.send("kafka-pay",jsonString);
        return R.OK();
    }
    @KafkaListener(topics = "kafka-pay")
    public void kafkaPay(
        String jsonString,Acknowledgment ack)  {
        //通过userId获取用户信息
        Order order = JSON.parseObject(jsonString, Order.class);
        Integer userId = order.getUserId();
        R userById = feignOrderService.getUserById(userId);
        String jsonString1 = JSON.toJSONString(userById.getData());
        TUser tUser1 = JSON.parseObject(jsonString1, TUser.class);
        String userPhone = tUser1.getUserPhone();
        String userEmeil = tUser1.getUserEmeil();
        if(order.getOrderStatus().equals(OrderEnum.ORDER_STATUS_SUCCESS.getCode())){
            MobileMessageUtil.sendMobileMessage(userPhone,"123456");
        }
        if(order.getOrderStatus().equals(OrderEnum.ORDER_STATUS_NO.getCode())){
            String content = order.getContent();
            MimeMessage mimeMessage =  javaMailSender.createMimeMessage();
            try {
                MimeMessageHelper mimeMessageHelper = new MimeMessageHelper(mimeMessage, true);
                mimeMessageHelper.setFrom("1683630933@qq.com");
                mimeMessageHelper.setTo(userEmeil);
                mimeMessageHelper.setSubject("您的订单已取消");
                mimeMessageHelper.setText("尊敬的"+tUser1.getUserName()+
                        ":您的订单已经取消，订单状态为:"+order.getOrderStatus()+
                        "取消原因:"+content);
                javaMailSender.send(mimeMessage);
            } catch (MessagingException e) {
                e.printStackTrace();
            }
        }
        ack.acknowledge();
    }
    @Override
    public R deleteOrder(Integer orderId) {
        Order order = orderMapper.selectById(orderId);
        if(ObjectUtils.isEmpty(order)){
            return R.ERROR(500,"数据不存在");
        }
        orderMapper.deleteById(orderId);
        return R.OK();
    }

    @Override
    public R deleteAll(List<Order> orderList) {
        for (Order order : orderList) {
            orderMapper.deleteById(order.getId());
        }
        return R.OK();
    }

    @Override
    public R sendMsg(String msg) {
        System.out.println("生产者发送的消息:"+msg);
        kafkaTemplate.send("kafka-msg",msg);
        return R.OK();
    }

    /**
     * 消费者
     * @param msg
     * @param ack
     */
    @KafkaListener(topics = "kafka-msg")
    public void consumMsg(String msg, Acknowledgment ack){
        System.out.println("消费者接收到的消息:"+msg);
        ack.acknowledge();
    }
}
